#ifndef HASHBOX_H_JM_20120910
#define HASHBOX_H_JM_20120910
/** @file HashBox.h
 * _OK_ Box obstaravajuci zahashovanie jednej relacie do hashovacej tabulky.
 **/

// ##
#include "../RootBox.h"
#include "../../common_bbx/sharedBoxData/HashTableHolder.h"
#include "../../LogConf.h"
// ## bobox
#include <bobox_box.hpp>
#include <bobox_types.hpp>

#include "../BoxProfiler.h"

namespace jm4bbx {

	/**
	 * HashBox prijma na vstupe data a pripadne nastavene vstupne signaly
	 * z definicie v modele vypoctu. Data nasledne zahashuje do zdielanej 
	 * hashovacej tabulky. Po ukonceni prace rozosle nastavene vystupne 
	 * sighaly podla urcenia v modele vypoctu.
	 *
	 * @tparam policyT typ politiky hashovanych data (spatna kompatibilita)
	 * @tparam valueT typ hodnoty hashovanych dat
	 * @tparam TABLE_SIZE velkost tabulky do ktorej sa bude hashovat
	 **/
	template<class policyT, typename keyT>
	class  HashBox : public RootBox {
		JM_USE_LOGGER(boxes_hash_hashBox_logger)
		JM_USE_EXPLICIT_LOGGER(Step1Tracer)
	public:
		/// Vstupna skupina pre data.
		static const  groupType INPUT_GROUP;
		/// Vtsyupny dtlpec pre data.
		static const  columnType INPUT_COLUMN;
		/// Vstupny parameter v definicii modelu.
		static const  inType IN_ARC;	
		/// Velkost hashovacej tabulky.
		//static const size_t HASH_TABLE_SIZE = TABLE_SIZE;
	
		/// Typ politiky hashovanych dat - spatna kompatibilita boboxu.
		typedef policyT policyType;
		/// Typ prenasanych dat medzi boxami.
		typedef KeyRecord<keyT> transferValueType;
		/// Typ holdra tabulky.
		typedef HashTableHolder<keyT> HTH;

	
		HashBox(const BoxParamPack& p) : RootBox(p) { }

		virtual void initImpl() {
			add_group(INPUT_GROUP, INPUT_COLUMN, get_input_descriptors(IN_ARC));
			wait4Receive();
		}

		virtual void syncMachEtwas() {
			JM_BOX_PROFILER_START

			GroupPtr ig = getGroupPtr(INPUT_GROUP);
			//if (ig->isPoisoned() || areSignalsInPoisoned()) {				
			//	send_poisoned(OUT_ARC);			
			//	sendSignalsOut(true);
			//	wait4Receive();
			//	return;
			//}

			// Zapis do tabulky -> nutna synchronizacia.
			HTH::getHashTable()->setSynchronized(true);

			LOG_I("Pocet {hashovanych} zaznamov od aktualnej pozicie po koniec: " 
				<< ig->getReadEnd() - ig->getReadPosition());
			LOG_I("Pocet {hashovanych} zaznamov od pociatku po koniec: " 
				<< ig->getReadEnd() - ig->getReadBegin());

			for (; ig->getReadPosition() < ig->getReadEnd(); ig->moveReadPosition()) {								
				// hashovanie
				transferValueType value = 
					get_data<policyType>(INPUT_COLUMN)[ig->getReadPosition()];				
				
				auto key = value.getKey();

				if (!HTH::getHashTable()->addIf(key, 
					HTH::ValueType(value.getIndex(), key), 
					jmlib::HashRowActionCondition::HAS_NOT_KEY)) {						
						HTH::getHashTable()->applyOnExtended(key, 
							[&](HTH::ExtendedHashEntry& entry){
								// Pri rovnosti klucov sa nevklada do mapy ale sa 
								// len prida index -> pocitadlo zaznamov s rovnakym 
								// klucom musi byt vzdy 1
								if (entry.getCount() != 1) {
									LOG_E("entry.count != 1 ale entry.count == " 
										<< entry.getCount());
								}
								assert(1 == entry.getCount());							
								entry.plainEntry.value.insert(value.getIndex());
						});
				}

			}

			LOG_I("Koniec hashovania.");

			sendSignalsOut();
			wait4Receive();
			JM_BOX_PROFILER_STOP(Step1Tracer)
		}
	protected:
		/// Zahajenie cakania na vstupne data a pripadne vstupene signaly.
		void wait4Receive() {
			wait4SignalsIn();
			receive(IN_ARC, INPUT_GROUP);
		}
	};
	

	template<class X, typename Y>
	const groupType  HashBox<X, Y>::INPUT_GROUP = groupType(0);

	template<class X, typename Y>
	const columnType  HashBox<X, Y>::INPUT_COLUMN = columnType(0);

	template<class X, typename Y>
	const inType  HashBox<X, Y>::IN_ARC = inType(0);


	typedef HashBox<IntegerKeyRecordPolicyType, integerType> IntegerHashBox;
	typedef HashBox<StringKeyRecordPolicyType, stringType> StringHashBox;

} // namespace jm4bbx


#endif
